package com.myself.storm_test;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import backtype.storm.Config;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
/**
 * 对字符串进行分隔发送
 */
@SuppressWarnings("serial")
public class SplitSentenceBolt extends BaseBasicBolt{

	Log log = LogFactory.getLog(SplitSentenceBolt.class);
	
	@SuppressWarnings("unchecked")
	public void prepare(Map stormConf, TopologyContext context) {
	}

	public void execute(Tuple tuple, BasicOutputCollector collector) {
		if (TupleHelpers.isTickTuple(tuple)) {
			System.out.println("每3秒定时输出!!!");
			return;
		}
		String message = tuple.getString(0);
		System.out.println(message);
		if (message == null || "".equals(message)) {
			return;
		}
		String[] word = message.split(" ");
		for(String value : word){
			collector.emit(new Values(value));
		}
	}
	
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		//
		declarer.declare(new Fields("word"));
	}

	public Map<String, Object> getComponentConfiguration() {
		Map<String, Object> conf = new HashMap<String, Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 3);
		return conf;
	}
}
